"""
@File    :   service.py
@Time    :   2024/02/17 22:29:09
@Author  :   Qrj
@Email   :   923207736@qq.com
@description   :   用户相关业务逻辑
"""

from datetime import datetime
from ipaddress import AddressValueError, IPv4Address, IPv4Network

from cashews import NOT_NONE
from sqlalchemy import and_, delete, func, insert, or_, select, update
from sqlalchemy.ext.asyncio import AsyncSession

from src.business.admin.system.menu.schemas.out import ListMenuCompleteOutSchema
from src.business.admin.system.role.models import Role
from src.business.admin.system.role.service import RoleService
from src.business.admin.system.settings.enums import PasswordStrengthEnum
from src.business.admin.system.settings.utils.password_strength import (
    PasswordStrengthUtil,
)
from src.business.admin.system.tenant.package.service import TenantPackageService
from src.business.admin.system.tenant.tenant.service import TenantService
from src.business.admin.system.user.schemas.out import (
    ListUserCompleteOutSchema,
    UserCompleteAndHavePasswordHashOutSchema,
    UserCompleteOutSchema,
    UserSelfAndMenuOutSchema,
)
from src.business.admin.system.user.schemas.validate import (
    UserAssignRoleValidateSchema,
    UserCreateValidateSchema,
    UserDeleteValidateSchema,
    UserListSearchValidateSchema,
    UserResetPasswordValidateSchema,
    UserSelfUpdateAvatarValidateSchema,
    UserSelfUpdateInfoValidateSchema,
    UserSelfUpdatePasswordValidateSchema,
    UserUpdateStatusValidateSchema,
    UserUpdateValidateSchema,
)
from src.exceptions.custom_async_validator import generate_async_validator_exception
from src.schemas.validate import (
    OrderBySearchValidateSchema,
    PaginationSearchValidateSchema,
)
from src.utils.cashews import CashewsCondition, data_cache
from src.utils.list_to_tree import ListConvertToTreeUtil
from src.utils.password_encryption import PswHashUtil

from .models import User, UserRoleRel


class UserService:
    """用户业务逻辑类"""

    @data_cache(
        ttl="30m",
        key="user#id:{tenant_id}:{user_id}:{need_password_hash}",
        tags=[
            "user#id:{tenant_id}",
            "user#id:{tenant_id}:{user_id}",
            "user#id:{tenant_id}:{user_id}:{need_password_hash}",
        ],
        condition=NOT_NONE,
    )
    @staticmethod
    async def get_user_use_id(
        db_session: AsyncSession,
        tenant_id: int,
        user_id: int,
        need_password_hash: bool = False,
        raise_not_exists_exception: bool = True,
        raise_disabled_exception: bool = True,
    ) -> UserCompleteOutSchema | UserCompleteAndHavePasswordHashOutSchema | None:
        """查找用户(用户ID)

        Args:
            db_session (AsyncSession): 数据库连接池
            tenant_id (int): 租户ID
            user_id (int): 用户ID
            need_password_hash: 是否需要密碼hash
            raise_not_exists_exception (bool, optional): 未找到用户时是否报错
            raise_disabled_exception (bool, optional): 禁用时是否报错

        Returns:
            UserCompleteOutSchema | UserCompleteAndHavePasswordHashOutSchema | None: 用户信息
        """
        out_schema = (
            UserCompleteAndHavePasswordHashOutSchema
            if need_password_hash
            else UserCompleteOutSchema
        )
        search_sql = select(*out_schema.convert_to_sql_model_field(User)).where(
            User.id == user_id, User.tenant_id == tenant_id
        )
        async with db_session.begin():
            search_res = await db_session.execute(search_sql)
        result = search_res.mappings().first()
        if result is not None:
            # 转换成pydantic的序列化模型
            result = out_schema.model_validate(result)
            if (not result.status) and raise_disabled_exception:
                # 禁用则报错
                raise generate_async_validator_exception("id", "用户已被禁用")
            return result
        elif result is None and raise_not_exists_exception:
            # 未找到则报错
            raise generate_async_validator_exception("id", "用户不存在")
        else:
            return result

    @data_cache(
        ttl="30m",
        key="user#username:{tenant_id}:{username}:{need_password_hash}",
        tags=[
            "user#username:{tenant_id}",
            "user#username:{tenant_id}:{username}",
            "user#username:{tenant_id}:{username}:{need_password_hash}",
        ],
        condition=NOT_NONE,
    )
    @staticmethod
    async def get_user_use_username(
        db_session: AsyncSession,
        tenant_id: int,
        username: str,
        need_password_hash: bool = False,
        raise_not_exists_exception: bool = True,
        raise_disabled_exception: bool = True,
    ) -> UserCompleteOutSchema | UserCompleteAndHavePasswordHashOutSchema | None:
        """查找用户(用户名)

        Args:
            db_session (AsyncSession): 数据库连接池
            tenant_id (int): 租户ID
            username (str): 用户名
            need_password_hash: 是否需要密碼hash
            raise_not_exists_exception (bool, optional): 未找到时是否报错
            raise_disabled_exception (bool, optional): 禁用时是否报错

        Returns:
            UserCompleteOutSchema | UserCompleteAndHavePasswordHashOutSchema | None: 用户信息
        """
        out_schema = (
            UserCompleteAndHavePasswordHashOutSchema
            if need_password_hash
            else UserCompleteOutSchema
        )
        search_sql = select(*out_schema.convert_to_sql_model_field(User)).where(
            User.username == username, User.tenant_id == tenant_id
        )
        async with db_session.begin():
            search_res = await db_session.execute(search_sql)
        result = search_res.mappings().first()
        if result is not None:
            # 转换成pydantic的序列化模型
            result = out_schema.model_validate(result)
            if (not result.status) and raise_disabled_exception:
                # 禁用则报错
                raise generate_async_validator_exception("username", "用户已被禁用")
            return result
        elif result is None and raise_not_exists_exception:
            # 未找到则报错
            raise generate_async_validator_exception("username", "用户不存在")
        else:
            return result

    @data_cache(
        ttl="30m",
        key="user#phone:{tenant_id}:{phone}:{need_password_hash}",
        tags=[
            "user#phone:{tenant_id}",
            "user#phone:{tenant_id}:{phone}",
            "user#phone:{tenant_id}:{phone}:{need_password_hash}",
        ],
        condition=NOT_NONE,
    )
    @staticmethod
    async def get_user_use_phone(
        db_session: AsyncSession,
        tenant_id: int,
        phone: str,
        need_password_hash: bool = False,
        raise_not_exists_exception: bool = True,
        raise_disabled_exception: bool = True,
    ) -> UserCompleteOutSchema | UserCompleteAndHavePasswordHashOutSchema | None:
        """查找用户(电话)

        Args:
            db_session (AsyncSession): 数据库连接池
            tenant_id (int): 租户ID
            phone (str): 电话
            need_password_hash: 是否需要密碼hash
            raise_not_exists_exception (bool, optional): 未找到用户时是否报错
            raise_disabled_exception (bool, optional): 禁用时是否报错

        Returns:
            UserCompleteOutSchema | UserCompleteAndHavePasswordHashOutSchema | None: 用户信息
        """
        if phone:
            out_schema = (
                UserCompleteAndHavePasswordHashOutSchema
                if need_password_hash
                else UserCompleteOutSchema
            )
            search_sql = select(*out_schema.convert_to_sql_model_field(User)).where(
                User.phone == phone, User.tenant_id == tenant_id
            )
            async with db_session.begin():
                search_res = await db_session.execute(search_sql)
            result = search_res.mappings().first()
            if result is not None:
                # 转换成pydantic的序列化模型
                result = out_schema.model_validate(result)
                if (not result.status) and raise_disabled_exception:
                    # 禁用则报错
                    raise generate_async_validator_exception("phone", "用户已被禁用")
                return result
            elif result is None and raise_not_exists_exception:
                # 未找到则报错
                raise generate_async_validator_exception("phone", "用户不存在")
            else:
                return result

    @data_cache(
        ttl="30m",
        key="user#email:{tenant_id}:{email}:{need_password_hash}",
        tags=[
            "user#email:{tenant_id}",
            "user#email:{tenant_id}:{email}",
            "user#email:{tenant_id}:{email}:{need_password_hash}",
        ],
        condition=NOT_NONE,
    )
    @staticmethod
    async def get_user_use_email(
        db_session: AsyncSession,
        tenant_id: int,
        email: str,
        need_password_hash: bool = False,
        raise_not_exists_exception: bool = True,
        raise_disabled_exception: bool = True,
    ) -> UserCompleteOutSchema | UserCompleteAndHavePasswordHashOutSchema | None:
        """查找用户(邮箱)

        Args:
            db_session (AsyncSession): 数据库连接池
            tenant_id (int): 租户ID
            email (str): 邮箱
            need_password_hash: 是否需要密碼hash
            raise_not_exists_exception (bool, optional): 未找到用户时是否报错
            raise_disabled_exception (bool, optional): 禁用时是否报错

        Returns:
            UserCompleteOutSchema | UserCompleteAndHavePasswordHashOutSchema | None: 用户信息
        """
        if email:
            out_schema = (
                UserCompleteAndHavePasswordHashOutSchema
                if need_password_hash
                else UserCompleteOutSchema
            )
            search_sql = select(*out_schema.convert_to_sql_model_field(User)).where(
                User.email == email, User.tenant_id == tenant_id
            )
            async with db_session.begin():
                search_res = await db_session.execute(search_sql)
            result = search_res.mappings().first()
            if result is not None:
                # 转换成pydantic的序列化模型
                result = out_schema.model_validate(result)
                if (not result.status) and raise_disabled_exception:
                    # 禁用则报错
                    raise generate_async_validator_exception("email", "用户已被禁用")
                return result
            elif result is None and raise_not_exists_exception:
                # 未找到则报错
                raise generate_async_validator_exception("email", "用户不存在")
            else:
                return result

    @data_cache(
        ttl="3m",
        key="complete_user_list:{tenant_id}:{search_schema}:{pagination_schema}:{order_schema}",
        tags=["complete_user_list:{tenant_id}"],
        condition=CashewsCondition.condition_only_true_result,
    )
    @staticmethod
    async def list_search_user(
        db_session: AsyncSession,
        tenant_id: int,
        search_schema: UserListSearchValidateSchema,
        pagination_schema: PaginationSearchValidateSchema,
        order_schema: OrderBySearchValidateSchema,
        need_select_total: bool = True,
    ) -> tuple[list[UserCompleteOutSchema], int | None]:
        """完整信息的用户列表查询

        Args:
            db_session (AsyncSession): 数据库连接池
            tenant_id: 租户ID
            search_schema (MenuListSearchValidateSchema): 查询参数
            pagination_schema: 分页参数
            order_schema: 排序参数
            need_select_total: 是否需要查询总数

        Returns:
            list[MenuCompleteOutSchema]: 查询结果、查询总数
        """
        # 构建排序字段
        order_by_field_sql = order_schema.convert_to_order_by_sql(search_schema, User)
        # 构建查询sql
        list_search_sql = (
            select(*UserCompleteOutSchema.convert_to_sql_model_field(User))
            .where(
                and_(*search_schema.convert_to_where_sql(User)),
                User.tenant_id == tenant_id,
            )
            .order_by(order_by_field_sql)
            .offset(pagination_schema.offset)
            .limit(pagination_schema.limit)
        )
        # 开始查询
        async with db_session.begin():
            list_search_res = await db_session.execute(list_search_sql)
        result = list_search_res.mappings().all()
        # 根据need_select_total参数来确定是否需要查询总数
        if need_select_total:
            total_sql = (
                select(func.count())
                .select_from(User)
                .where(
                    and_(*search_schema.convert_to_where_sql(User)),
                    User.tenant_id == tenant_id,
                )
            )
            async with db_session.begin():
                total_res = await db_session.execute(total_sql)
            total = total_res.scalar()
        else:
            total = None
        # 统一转为pydantic列表模型
        return ListUserCompleteOutSchema.validate_python(result), total

    @data_cache(
        ttl="30m",
        key="user_role_ids:{tenant_id}:{user_id}",
        tags=["user_role_ids:{tenant_id}", "user_role_ids:{tenant_id}:{user_id}"],
        condition=NOT_NONE,
    )
    @staticmethod
    async def get_user_role_ids(
        db_session: AsyncSession, tenant_id: int, user_id: int
    ) -> list[int]:
        """获取用户的角色ID列表

        Args:
            db_session (AsyncSession): 数据库连接池
            tenant_id (int): 租户ID
            user_id (int): 用户ID

        Returns:
            list[int]: 用户的角色ID列表
        """
        # 验证用户是否存在
        await UserService.get_user_use_id(
            db_session, tenant_id, user_id, raise_disabled_exception=False
        )

        search_sql = select(UserRoleRel.role_id).where(
            UserRoleRel.user_id == user_id, UserRoleRel.tenant_id == tenant_id
        )
        async with db_session.begin():
            search_res = await db_session.execute(search_sql)
        result: list[int] = search_res.scalars().all()
        return result

    @staticmethod
    async def get_user_menu_id(
        db_session: AsyncSession, tenant_id: int, user_id: int
    ) -> list[int]:
        """获取用户的菜单ID列表(用于前端判断可见的菜单)

        Args:
            db_session (AsyncSession): 数据库连接池
            tenant_id (int): 租户ID
            user_id (int): 用户ID

        Returns:
            list[int]: 用户的菜单ID列表
        """
        # 验证用户是否存在
        user_obj = await UserService.get_user_use_id(
            db_session, tenant_id, user_id, raise_disabled_exception=False
        )
        # 管理员级别用户拥有租户套餐内部的所有菜单及其权限
        if user_obj.is_admin:
            # 获取租户对应的租户套餐
            tenant = await TenantService.get_tenant_use_id(
                db_session,
                tenant_id,
                raise_disabled_exception=False,
                raise_expired_exception=False,
            )
            user_menu_info_list = (
                await TenantPackageService.list_tenant_package_all_menu(
                    db_session, tenant.tenant_package_id
                )
            )
        else:
            # 获取用户的所有角色
            user_role_ids = await UserService.get_user_role_ids(
                db_session, tenant_id, user_id
            )
            # 获取用户所有角色的所有菜单信息
            user_menu_info_list = await RoleService.batch_get_role_menu_info(
                db_session, tenant_id, *user_role_ids
            )
        return [i.id for i in user_menu_info_list]

    @staticmethod
    async def get_user_menu_permission(
        db_session: AsyncSession, tenant_id: int, user_id: int
    ) -> list[str]:
        """获取用户的菜单权限列表(用于菜单权限验证)

        Args:
            db_session (AsyncSession): 数据库连接池
            tenant_id (int): 租户ID
            user_id (int): 用户ID

        Returns:
            list[str]: 用户的菜单权限列表
        """
        user_self_and_menu_info = await UserService.get_user_self_and_menu_info(
            db_session, tenant_id, user_id
        )
        return user_self_and_menu_info.menu_permissions

    @staticmethod
    async def get_user_self_and_menu_info(
        db_session: AsyncSession, tenant_id: int, user_id: int
    ) -> UserSelfAndMenuOutSchema:
        """获取用户本身及其菜单相关的信息

        Args:
            db_session (AsyncSession): 数据库连接池
            tenant_id (int): 租户ID
            user_id (int): 用户ID

        Returns:
            UserSelfAndMenuOutSchema: 用户本身及其菜单相关的信息
        """
        # 验证用户是否存在
        user_obj = await UserService.get_user_use_id(
            db_session, tenant_id, user_id, raise_disabled_exception=False
        )
        # 管理员级别用户拥有租户套餐内部的所有菜单及其权限
        if user_obj.is_admin:
            # 获取租户对应的租户套餐
            tenant = await TenantService.get_tenant_use_id(
                db_session,
                tenant_id,
                raise_disabled_exception=False,
                raise_expired_exception=False,
            )
            user_menu_info_list = (
                await TenantPackageService.list_tenant_package_all_menu(
                    db_session, tenant.tenant_package_id
                )
            )
        else:
            # 获取用户的所有角色
            user_role_ids = await UserService.get_user_role_ids(
                db_session, tenant_id, user_id
            )
            # 获取用户所有角色的所有菜单信息
            user_menu_info_list = await RoleService.batch_get_role_menu_info(
                db_session, tenant_id, *user_role_ids
            )
        # 将用户所有角色的所有菜单信息转为树形列表
        user_menu_info_tree_list = ListConvertToTreeUtil.convert(
            ListMenuCompleteOutSchema.dump_python(user_menu_info_list)
        )
        # 获取菜单权限标识列表
        menu_permissions = [
            i.permission for i in user_menu_info_list if i.permission != ""
        ]
        return UserSelfAndMenuOutSchema.model_validate(
            {
                "user": user_obj.model_dump(),
                "menus": user_menu_info_tree_list,
                "menu_permissions": menu_permissions,
            }
        )

    @staticmethod
    @data_cache.transaction()
    async def create_user(
        db_session: AsyncSession,
        tenant_id: int,
        create_schema: UserCreateValidateSchema,
        creator_username: str = "",
        is_admin: bool = False,
    ) -> int:
        """创建用户

        Args:
            db_session (AsyncSession): 数据库连接池
            tenant_id (int): 租户ID
            create_schema (UserCreateValidateSchema): 创建所需的信息
            creator_username: 创建者用户名
            is_admin: 是否是管理员用户

        Returns:
            int: 创建的sql的主键ID
        """
        # 验证用户的手机号、邮箱是否已在指定租户下存在(用户名的同名检测直接通过mysql的联合unique实现)
        await UserService.verify_phone_or_email_exists(
            db_session, tenant_id, phone=create_schema.phone, email=create_schema.email
        )
        # 验证密码强度
        await UserService.verify_password_strength(
            db_session, tenant_id, create_schema.password
        )
        # 获取密码hash
        password_hash = await PswHashUtil.async_hash_psw(create_schema.password)
        # 构建用户的创建信息字典(排除password参数)
        create_info_dict = create_schema.model_dump(exclude=("password",))
        # 创建
        create_sql = insert(User).values(
            tenant_id=tenant_id,
            password_hash=password_hash,
            creator_username=creator_username,
            is_admin=is_admin,
            **create_info_dict,
        )
        async with db_session.begin():
            create_res = await db_session.execute(create_sql)
            # 清空相关缓存标签
            await data_cache.delete_tags(
                f"complete_user_list:{tenant_id}",
            )
        return create_res.inserted_primary_key_rows[0][0]

    @staticmethod
    @data_cache.transaction()
    async def update_self_info(
        db_session: AsyncSession,
        tenant_id: int,
        self_id: int,
        update_self_schema: UserSelfUpdateInfoValidateSchema,
    ):
        """用户修改自身信息

        Args:
            db_session (AsyncSession): 数据库连接池
            tenant_id (int): 租户ID
            self_id (int): 当前用户ID
            update_self_schema (UserSelfUpdateInfoValidateSchema): 修改所需数据
        """
        # 获取当前用户实例
        self_user_obj = await UserService.get_user_use_id(
            db_session, tenant_id, self_id, raise_disabled_exception=False
        )
        # 验证用户的手机号、邮箱是否已在指定租户下存在(用户名的同名检测直接通过mysql的联合unique实现)
        await UserService.verify_phone_or_email_exists(
            db_session,
            tenant_id,
            phone=update_self_schema.phone,
            email=update_self_schema.email,
            current_id=self_id,
        )
        # 构建修改所需的数据(排除未设置的值和id)
        update_schema_data = update_self_schema.model_dump(exclude_unset=True)
        if update_schema_data:
            # 修改sql
            update_sql = (
                update(User)
                .values(
                    **update_schema_data,
                    self_info_last_modified_datetime=datetime.now(),
                )
                .where(User.id == self_id, User.tenant_id == tenant_id)
            )
            # 修改操作
            async with db_session.begin():
                # 修改sql数据
                await db_session.execute(update_sql)
                # 清空相关缓存标签
                await data_cache.delete_tags(
                    f"user#id:{tenant_id}:{self_id}",
                    f"user#username:{tenant_id}:{self_user_obj.username}",
                    f"user#phone:{tenant_id}:{self_user_obj.phone}",
                    f"user#email:{tenant_id}:{self_user_obj.email}",
                    f"complete_user_list:{tenant_id}",
                )

    @staticmethod
    @data_cache.transaction()
    async def update_self_avatar(
        db_session: AsyncSession,
        tenant_id: int,
        self_id: int,
        update_self_schema: UserSelfUpdateAvatarValidateSchema,
    ):
        """用户修改自身头像

        Args:
            db_session (AsyncSession): 数据库连接池
            tenant_id (int): 租户ID
            self_id (int): 当前用户ID
            update_self_schema (UserSelfUpdateAvatarValidateSchema): 修改所需数据
        """
        # 获取当前用户实例
        self_user_obj = await UserService.get_user_use_id(
            db_session, tenant_id, self_id, raise_disabled_exception=False
        )
        # 修改sql
        update_sql = (
            update(User)
            .values(
                avatar=update_self_schema.avatar,
                self_info_last_modified_datetime=datetime.now(),
            )
            .where(User.id == self_id, User.tenant_id == tenant_id)
        )
        # 修改操作
        async with db_session.begin():
            # 修改sql数据
            await db_session.execute(update_sql)
            # 清空相关缓存标签
            await data_cache.delete_tags(
                f"user#id:{tenant_id}:{self_id}",
                f"user#username:{tenant_id}:{self_user_obj.username}",
                f"user#phone:{tenant_id}:{self_user_obj.phone}",
                f"user#email:{tenant_id}:{self_user_obj.email}",
                f"complete_user_list:{tenant_id}",
            )

    @staticmethod
    async def update_self_password(
        db_session: AsyncSession,
        tenant_id: int,
        self_id: int,
        update_self_schema: UserSelfUpdatePasswordValidateSchema,
    ):
        """用户修改自身密码

        Args:
            db_session (AsyncSession): 数据库连接池
            tenant_id (int): 租户ID
            self_id (int): 当前用户ID
            update_self_schema (UserSelfUpdatePasswordValidateSchema): 修改所需数据
        """
        # 获取当前用户实例
        self_user_obj = await UserService.get_user_use_id(
            db_session, tenant_id, self_id, raise_disabled_exception=False
        )
        # 验证密码强度
        await UserService.verify_password_strength(
            db_session, tenant_id, update_self_schema.password
        )
        # 获取密码hash
        password_hash = await PswHashUtil.async_hash_psw(update_self_schema.password)
        # 修改sql
        update_sql = (
            update(User)
            .values(
                password_hash=password_hash,
                self_psw_last_modified_datetime=datetime.now(),
            )
            .where(User.id == self_id, User.tenant_id == tenant_id)
        )
        # 修改操作
        async with db_session.begin():
            # 修改sql数据
            await db_session.execute(update_sql)
            # 清空相关缓存标签
            await data_cache.delete_tags(
                f"user#id:{tenant_id}:{self_id}",
                f"user#username:{tenant_id}:{self_user_obj.username}",
                f"user#phone:{tenant_id}:{self_user_obj.phone}",
                f"user#email:{tenant_id}:{self_user_obj.email}",
                f"complete_user_list:{tenant_id}",
            )

    @staticmethod
    @data_cache.transaction()
    async def update_user(
        db_session: AsyncSession,
        tenant_id: int,
        update_schema: UserUpdateValidateSchema,
        updater_username: str = "",
    ):
        """修改用户信息

        Args:
            db_session (AsyncSession): 数据库连接池
            tenant_id (int): 租户ID
            update_schema (UserUpdateValidateSchema): 修改操作所需的数据
            updater_username (str, optional):修改者用户名
        """
        # 检查需修改的sql实例是否存在(禁用不报错)
        update_sql_obj = await UserService.get_user_use_id(
            db_session, tenant_id, update_schema.id, raise_disabled_exception=False
        )
        # 检查是否修改了管理员用户
        if update_sql_obj.is_admin:
            raise generate_async_validator_exception("id", "不能修改管理员用户信息")
        # 验证用户的手机号、邮箱是否已在指定租户下存在(用户名的同名检测直接通过mysql的联合unique实现)
        await UserService.verify_phone_or_email_exists(
            db_session,
            tenant_id,
            phone=update_schema.phone,
            email=update_schema.email,
            current_id=update_schema.id,
        )
        # 构建修改所需的数据(排除未设置的值和id)
        update_schema_data = update_schema.model_dump(
            exclude=("id",), exclude_unset=True
        )
        if update_schema_data:
            # 修改sql
            update_sql = (
                update(User)
                .values(
                    **update_schema_data,
                    updater_username=updater_username,
                    admin_last_modified_datetime=datetime.now(),
                )
                .where(User.id == update_schema.id, User.tenant_id == tenant_id)
            )
            # 修改操作
            async with db_session.begin():
                # 修改sql数据
                await db_session.execute(update_sql)
                # 清空相关缓存标签
                await data_cache.delete_tags(
                    f"user#id:{tenant_id}:{update_sql_obj.id}",
                    f"user#username:{tenant_id}:{update_sql_obj.username}",
                    f"user#phone:{tenant_id}:{update_sql_obj.phone}",
                    f"user#email:{tenant_id}:{update_sql_obj.email}",
                    f"complete_user_list:{tenant_id}",
                )

    @staticmethod
    @data_cache.transaction()
    async def update_user_status(
        db_session: AsyncSession,
        tenant_id: int,
        update_schema: UserUpdateStatusValidateSchema,
        updater_username: str = "",
    ):
        """修改用户状态

        Args:
            db_session (AsyncSession): 数据库连接池
            tenant_id (int): 租户ID
            update_schema (UserUpdateStatusValidateSchema): 修改操作所需的数据
            updater_username (str, optional):修改者用户名
        """
        # 检查需修改的sql实例是否存在(禁用不报错)
        update_sql_obj = await UserService.get_user_use_id(
            db_session, tenant_id, update_schema.id, raise_disabled_exception=False
        )
        # 检查是否修改了管理员用户
        if update_sql_obj.is_admin:
            raise generate_async_validator_exception("id", "不能修改管理员用户信息")
        # 修改sql
        update_sql = (
            update(User)
            .values(
                status=update_schema.status,
                updater_username=updater_username,
                admin_last_modified_datetime=datetime.now(),
            )
            .where(User.id == update_schema.id, User.tenant_id == tenant_id)
        )
        # 修改操作
        async with db_session.begin():
            # 修改sql数据
            await db_session.execute(update_sql)
            # 清空相关缓存标签
            await data_cache.delete_tags(
                f"user#id:{tenant_id}:{update_sql_obj.id}",
                f"user#username:{tenant_id}:{update_sql_obj.username}",
                f"user#phone:{tenant_id}:{update_sql_obj.phone}",
                f"user#email:{tenant_id}:{update_sql_obj.email}",
                f"complete_user_list:{tenant_id}",
            )

    @staticmethod
    async def reset_user_password(
        db_session: AsyncSession,
        tenant_id: int,
        reset_schema: UserResetPasswordValidateSchema,
        updater_username: str = "",
    ):
        """重置用户密码

        Args:
            db_session (AsyncSession): 数据库连接池
            tenant_id (int): 租户ID
            reset_schema (UserResetPasswordValidateSchema): 重置操作所需的数据
            updater_username (str, optional): 修改者用户名
        """
        # 查找需重置密码的sql实例(禁用不报错)
        reset_user = await UserService.get_user_use_id(
            db_session, tenant_id, reset_schema.id, raise_disabled_exception=False
        )
        # 检查需重置密码的用户是否是管理员用户
        if reset_user.is_admin:
            raise generate_async_validator_exception(
                "id", "只能通过服务器cli命令对系统级别用户进行密码重置操作"
            )
        # 验证密码强度
        await UserService.verify_password_strength(
            db_session, tenant_id, reset_schema.password
        )
        # 获取密码hash
        password_hash = await PswHashUtil.async_hash_psw(reset_schema.password)
        # 重置的sql
        reset_sql = (
            update(User)
            .values(
                password_hash=password_hash,
                updater_username=updater_username,
                admin_pwd_last_reset_datetime=datetime.now(),
            )
            .where(User.id == reset_schema.id, User.tenant_id == tenant_id)
        )
        # 重置操作
        async with db_session.begin():
            await db_session.execute(reset_sql)
            # 清空相关缓存
            await data_cache.delete_tags(
                f"user#id:{tenant_id}:{reset_user.id}",
                f"user#username:{tenant_id}:{reset_user.username}",
                f"user#phone:{tenant_id}:{reset_user.phone}",
                f"user#email:{tenant_id}:{reset_user.email}",
                f"complete_user_list:{tenant_id}",
            )

    @staticmethod
    @data_cache.transaction()
    async def assign_user_role(
        db_session: AsyncSession,
        tenant_id: int,
        assign_schema: UserAssignRoleValidateSchema,
    ):
        """分配用户角色

        Args:
            db_session (AsyncSession): 数据库连接池
            tenant_id (int): 租户ID
            assign_schema (UserAssignRoleValidateSchema): 分配操作所需的数据
        """
        # 查找需分配的sql实例(禁用不报错)
        assign_user = await UserService.get_user_use_id(
            db_session, tenant_id, assign_schema.id, raise_disabled_exception=False
        )
        # 检查需分配的用户是否是管理员用户
        if assign_user.is_admin:
            raise generate_async_validator_exception(
                "id", "不允许对管理员级别用户进行分配角色操作"
            )
        # 只留下实际存在的角色ID
        if assign_schema.role_ids:
            exists_role_ids_sql = select(Role.id).where(
                Role.id.in_(assign_schema.role_ids)
            )
            async with db_session.begin():
                exists_role_ids_res = await db_session.execute(exists_role_ids_sql)
            exists_role_ids = exists_role_ids_res.scalars().all()
        else:
            exists_role_ids = assign_schema.role_ids

        # 修改中间表(先删再添加)
        # 删除的sql
        delete_sql = delete(UserRoleRel).where(
            UserRoleRel.user_id == assign_schema.id, UserRoleRel.tenant_id == tenant_id
        )
        # 批量新增的sql
        insert_batch_dict_list = [
            {"tenant_id": tenant_id, "role_id": role_id, "user_id": assign_schema.id}
            for role_id in exists_role_ids
        ]
        insert_sql = (insert(UserRoleRel), insert_batch_dict_list)
        # 整体操作
        async with db_session.begin():
            # 删除操作
            await db_session.execute(delete_sql)
            # 批量新增操作(非空才需要添加)
            if insert_batch_dict_list:
                await db_session.execute(*insert_sql)
            # 清空相关缓存标签
            await data_cache.delete_tags(
                f"user_role_ids:{tenant_id}:{assign_schema.id}"
            )

    @staticmethod
    @data_cache.transaction()
    async def delete_user(
        db_session: AsyncSession,
        tenant_id: int,
        delete_schema: UserDeleteValidateSchema,
    ):
        """删除用户

        Args:
            db_session (AsyncSession): 数据库连接池
            delete_schema (UserDeleteValidateSchema): 删除操作所属的数据
        """
        # 查找需删除的用户
        # 构建查询sql
        select_sql = select(
            User.id, User.is_admin, User.email, User.username, User.phone
        ).where(User.id.in_(delete_schema.ids), User.tenant_id == tenant_id)
        # 进行sql查询
        async with db_session.begin():
            select_res = await db_session.execute(select_sql)
        select_result = select_res.mappings().all()
        # 检查是否删除了管理员级别用户
        if any([i["is_admin"] for i in select_result]):
            raise generate_async_validator_exception("ids", "不允许删除管理员用户")

        # 进行删除操作
        if select_result:
            # 获取需要删除的用户id列表
            delete_id_list = [i["id"] for i in select_result]
            # 获取需要删除的用户名列表
            delete_username_list = [i["username"] for i in select_result]
            # 获取需要删除的用户手机号码列表(非空才删)
            delete_phone_list = [i["phone"] for i in select_result if i["phone"]]
            # 获取需要删除的用户手机号码列表(非空才删)
            delete_email_list = [i["email"] for i in select_result if i["email"]]
            # 删除用户本身的sql
            delete_sql = delete(User).where(User.id.in_(delete_id_list))
            # 删除用户的相关角色sql
            delete_user_role_rel_sql = delete(UserRoleRel).where(
                UserRoleRel.user_id.in_(delete_id_list)
            )
            # 删除操作
            async with db_session.begin():
                # 删除sql中的数据
                await db_session.execute(delete_sql)
                await db_session.execute(delete_user_role_rel_sql)
                delete_cache_tags = [
                    *[f"user#id:{tenant_id}:{i}" for i in delete_id_list],
                    *[f"user#username:{tenant_id}:{i}" for i in delete_username_list],
                    *[f"user#phone:{tenant_id}:{i}" for i in delete_phone_list],
                    *[f"user#email:{tenant_id}:{i}" for i in delete_email_list],
                    *[f"user_role_ids:{tenant_id}:{i}" for i in delete_id_list],
                    f"complete_user_list:{tenant_id}",
                ]
                # 清空相关缓存标签
                await data_cache.delete_tags(*delete_cache_tags)

    @staticmethod
    async def verify_phone_or_email_exists(
        db_session: AsyncSession,
        tenant_id: int,
        phone: str = "",
        email: str = "",
        current_id: int = 0,
    ):
        """验证用户的手机号、邮箱是否已在指定租户下存在

        Args:
            db_session (AsyncSession): 数据库连接池
            tenant_id (int): 租户ID
            phone (str, optional): 手机号码
            email (str, optional): 邮箱
            current_id (int, optional): 当前ID, 修改时传入
        """
        # 构建需要查询的字段
        verify_sql = select(User.id, User.phone, User.email)
        # 分辨需要查找的字段
        verify_list = []
        if phone:
            verify_list.append(User.phone == phone)
        if email:
            verify_list.append(User.email == email)
        verify_list_len = len(verify_list)
        # 进行统一判断与查找
        if verify_list_len == 1:
            verify_sql = verify_sql.where(User.tenant_id == tenant_id, *verify_list)
        elif verify_list_len > 1:
            verify_sql = verify_sql.where(
                and_(or_(*verify_list), User.tenant_id == tenant_id)
            )
        else:
            # 没有传入参数则直接通过验证
            return
        # 进行sql查询
        async with db_session.begin():
            verify_res = await db_session.execute(verify_sql)
        result_list = verify_res.mappings().all()
        for result in result_list:
            # 有找到且非本id则直接报错
            if result["id"] != current_id:
                if result["phone"] == phone:
                    raise generate_async_validator_exception(
                        "phone", "用户的手机号码已存在"
                    )
                else:
                    raise generate_async_validator_exception(
                        "email", "用户的邮箱已存在"
                    )

    @staticmethod
    async def verify_password_strength(
        db_session: AsyncSession,
        tenant_id: int,
        password: str,
        column_name: str = "password",
    ):
        """验证密码是否符合当前租户的密码强度要求

        Args:
            db_session (AsyncSession): 数据库连接池
            tenant_id (int): 租户ID
            password (str): 密码
            column_name: 报错时使用的字段名
        """
        from src.business.admin.system.settings.service import SystemSettingsService

        # 获取租户对应的系统配置
        tenant_settings = await SystemSettingsService.get_system_settings_use_tenant_id(
            db_session, tenant_id
        )
        # 验证密码强度是否符合要求
        verify_result, err_msg = PasswordStrengthUtil.verify_password_strength(
            PasswordStrengthEnum(tenant_settings.password_strength),
            tenant_settings.min_password_length,
            tenant_settings.max_password_length,
            password,
        )
        if not verify_result:
            raise generate_async_validator_exception(column_name, err_msg)

    @staticmethod
    def validate_ip_can_login(
        login_ip: str,
        user_can_login_networks: list[IPv4Network],
        field_name: str = "login_ip",
    ) -> bool:
        """验证ip是否在用户的可登录IP段内

        Args:
            login_ip (str): 登录IP
            user_can_login_networks: 用户的可登录IP段内
            field_name (str): 报错的字段名称

        Returns:
            bool: ip是否在用户的可登录IP段内
        """
        # 当允许登录的IP段列表为空时, 表示允许任意IP登录
        if not user_can_login_networks:
            return True
        # 开始判断
        try:
            # 尝试将登录IP转为IPv4Address(转换失败则直接返回False)
            login_ip_obj = IPv4Address(login_ip)
            # 登录IP在任意一个可登录IP段内都表示验证通过
            for user_can_login_network in user_can_login_networks:
                if login_ip_obj in user_can_login_network:
                    return True
            raise generate_async_validator_exception(
                field_name, "IP不在用户可登录IP段内"
            )
        except (AddressValueError, ValueError):
            raise generate_async_validator_exception(
                field_name, "IP不在用户可登录IP段内"
            )
